import asyncio
import json
import logging
import os
import ssl
from typing import Any, cast

import websockets

# 导入配置
from speech_convert.config.funasr_config import CONFIG
# 导入初始化模块
from speech_convert.server.server_initialization import initialize_server

# 配置日志级别和环境变量以减少不必要的输出
os.environ["MODELSCOPE_LOGGING_LEVEL"] = "ERROR"  # 设置ModelScope日志级别
logging.getLogger("modelscope").setLevel(logging.ERROR)  # 抑制ModelScope INFO日志
logging.getLogger("funasr").setLevel(logging.ERROR)  # 抑制FunASR日志
logging.getLogger("jieba").setLevel(logging.ERROR)  # 抑制jieba DEBUG日志

# 初始化服务器 - 参数解析、模型加载、管理器创建
args, speaker_manager, response_manager, model_manager, websocket_users = (
    initialize_server()
)


async def ws_reset(websocket: Any):
    print(f"🧹 正在重置 WebSocket 连接: {websocket.remote_address}")

    # 清理缓存
    if hasattr(websocket, "status_dict_asr_online"):
        websocket.status_dict_asr_online["cache"] = {}
        websocket.status_dict_asr_online["is_final"] = True
    if hasattr(websocket, "status_dict_vad"):
        websocket.status_dict_vad["cache"] = {}
        websocket.status_dict_vad["is_final"] = True
    if hasattr(websocket, "status_dict_punc"):
        websocket.status_dict_punc["cache"] = {}

    # 只在连接仍然打开时才关闭
    try:
        if hasattr(websocket, "closed") and not websocket.closed:
            await websocket.close()
        elif not hasattr(websocket, "closed"):
            # 对于没有closed属性的websocket类型，直接尝试关闭
            await websocket.close()
    except Exception as e:
        print(f"⚠️ 关闭 WebSocket 时出错: {e}")


async def clear_websocket():
    for websocket in websocket_users:
        await ws_reset(websocket)
    websocket_users.clear()


async def ws_serve(websocket: Any, path: str | None = None):
    frames: list[bytes] = []
    frames_asr: list[bytes] = []
    frames_asr_online: list[bytes] = []
    global websocket_users

    # 用户列表添加当前websocket
    websocket_users.add(websocket)
    # 离线ASR模型参数
    websocket.status_dict_asr = cast(dict[str, Any], {})
    # 在线ASR模型参数
    websocket.status_dict_asr_online = cast(
        dict[str, Any],
        {
            "cache": {},
            "is_final": False,
        },
    )
    # VAD模型参数
    websocket.status_dict_vad = cast(
        dict[str, Any],
        {
            "cache": {},
            "is_final": False,
        },
    )
    # 标点符号模型参数 - 使用增强配置
    websocket.status_dict_punc = cast(
        dict[str, Any],
        {
            "cache": {},
            "density": CONFIG.punctuation.PUNCTUATION_DENSITY,
            "cache_size": CONFIG.punctuation.CACHE_SIZE,
            "force_punc": CONFIG.punctuation.FORCE_PUNCTUATION,
            "keep_spaces": CONFIG.punctuation.KEEP_ORIGINAL_SPACES,
        },
    )

    websocket.chunk_interval = CONFIG.audio.DEFAULT_CHUNK_INTERVAL
    websocket.vad_pre_idx = 0
    speech_start = False
    speech_end_i = -1
    websocket.wav_name = CONFIG.processing.DEFAULT_WAV_NAME
    websocket.mode = CONFIG.processing.DEFAULT_MODE

    # 消息统计
    msg_stats = {"config": 0, "audio": 0, "control": 0}
    audio_bytes_total = 0

    print(
        f"🔗 新用户连接: {websocket.remote_address}, 当前连接数: {len(websocket_users)}",
        flush=True,
    )

    try:
        print(f"📥 开始监听消息: {websocket.remote_address}\n", flush=True)
        async for message in websocket:
            # 统计消息类型
            if isinstance(message, str):
                try:
                    msg_json = json.loads(message)
                    if "is_speaking" in msg_json or "chunk_size" in msg_json:
                        msg_stats["config"] += 1
                        print(
                            f"📋 配置消息 #{msg_stats['config']}: {len(message)}字节",
                            flush=True,
                        )
                    else:
                        msg_stats["control"] += 1
                        print(
                            f"🎛️ 控制消息 #{msg_stats['control']}: {len(message)}字节",
                            flush=True,
                        )
                except:
                    msg_stats["control"] += 1
                    print(
                        f"🎛️ 控制消息 #{msg_stats['control']}: {len(message)}字节",
                        flush=True,
                    )
            else:
                msg_stats["audio"] += 1
                audio_bytes_total += len(message)
                # 每N个音频包更新一次统计
                if msg_stats["audio"] % CONFIG.audio.AUDIO_STATS_INTERVAL == 0:
                    print(
                        f"🎵 音频数据: {msg_stats['audio']}包, 总计{audio_bytes_total:,}字节",
                        end="\r",
                        flush=True,
                    )
                elif msg_stats["audio"] == 1:
                    print(
                        f"🎵 音频数据: {msg_stats['audio']}包, 总计{audio_bytes_total:,}字节",
                        end="\r",
                        flush=True,
                    )
            if isinstance(message, str):
                messagejson = json.loads(message)

                if "is_speaking" in messagejson:
                    websocket.is_speaking = messagejson["is_speaking"]
                    websocket.status_dict_asr_online["is_final"] = (
                        not websocket.is_speaking
                    )
                if "chunk_interval" in messagejson:
                    websocket.chunk_interval = messagejson["chunk_interval"]
                if "wav_name" in messagejson:
                    websocket.wav_name = messagejson.get("wav_name")
                if "chunk_size" in messagejson:
                    chunk_size = messagejson["chunk_size"]
                    if isinstance(chunk_size, str):
                        chunk_size = chunk_size.split(",")
                    websocket.status_dict_asr_online["chunk_size"] = [
                        int(x) for x in chunk_size
                    ]
                if "encoder_chunk_look_back" in messagejson:
                    websocket.status_dict_asr_online["encoder_chunk_look_back"] = (
                        messagejson["encoder_chunk_look_back"]
                    )
                if "decoder_chunk_look_back" in messagejson:
                    websocket.status_dict_asr_online["decoder_chunk_look_back"] = (
                        messagejson["decoder_chunk_look_back"]
                    )
                if "hotwords" in messagejson:
                    websocket.status_dict_asr["hotword"] = messagejson["hotwords"]
                if "mode" in messagejson:
                    websocket.mode = messagejson["mode"]

            # 初始化chunk_size（如果还没有设置）
            if "chunk_size" not in websocket.status_dict_asr_online:
                websocket.status_dict_asr_online["chunk_size"] = [5, 10]

            websocket.status_dict_vad["chunk_size"] = int(
                websocket.status_dict_asr_online["chunk_size"][1]
                * CONFIG.audio.VAD_CHUNK_MULTIPLIER
                / websocket.chunk_interval
            )
            if (
                len(frames_asr_online) > 0
                or len(frames_asr) >= 0
                or not isinstance(message, str)
            ):
                if not isinstance(message, str):
                    frames.append(message)
                    duration_ms = len(message) // CONFIG.audio.AUDIO_DURATION_DIVISOR
                    websocket.vad_pre_idx += duration_ms

                    # asr online
                    frames_asr_online.append(message)
                    websocket.status_dict_asr_online["is_final"] = speech_end_i != -1
                    if (
                        len(frames_asr_online) % websocket.chunk_interval == 0
                        or websocket.status_dict_asr_online["is_final"]
                    ):
                        if (
                            websocket.mode == CONFIG.processing.MODE_2PASS
                            or websocket.mode == CONFIG.processing.MODE_ONLINE
                        ):
                            audio_in = b"".join(frames_asr_online)
                            try:
                                await async_asr_online(websocket, audio_in)
                            except:
                                # 这里通常是客户端断开连接，所以后续的数据包任务可以不用处理，直接跳过
                                # print(f"error in asr streaming, {websocket.status_dict_asr_online}")
                                pass
                        frames_asr_online = []
                    if speech_start:
                        frames_asr.append(message)
                    # vad online
                    speech_start_i, speech_end_i = -1, -1
                    try:
                        speech_start_i, speech_end_i = await async_vad(
                            websocket, message
                        )
                    except:
                        print("error in vad")
                    if speech_start_i != -1:
                        speech_start = True
                        beg_bias = (
                            websocket.vad_pre_idx - speech_start_i
                        ) // duration_ms
                        frames_pre = frames[-beg_bias:]
                        frames_asr = []
                        frames_asr.extend(frames_pre)
                # asr punc offline
                if speech_end_i != -1 or not websocket.is_speaking:
                    # print("vad end point")
                    if (
                        websocket.mode == CONFIG.processing.MODE_2PASS
                        or websocket.mode == CONFIG.processing.MODE_OFFLINE
                    ):
                        audio_in = b"".join(frames_asr)
                        try:
                            await async_asr(websocket, audio_in)
                        except Exception as e:
                            print(f"❌ ASR离线处理错误: {e}")
                            # 可选：发送错误响应给客户端
                            # error_response = response_manager.create_error_response(f"ASR处理失败: {str(e)}")
                            # await websocket.send(json.dumps(error_response.to_dict()))
                    frames_asr = []
                    speech_start = False
                    frames_asr_online = []
                    websocket.status_dict_asr_online["cache"] = {}
                    if not websocket.is_speaking:
                        websocket.vad_pre_idx = 0
                        frames = []
                        websocket.status_dict_vad["cache"] = {}
                    else:
                        frames = frames[-CONFIG.audio.FRAMES_CACHE_SIZE :]

        # 正常结束循环，说明客户端正常发送了关闭帧
        print(f"\n✅ 客户端正常断开连接: {websocket.remote_address}", flush=True)
        print(
            f"📊 会话统计 - 配置:{msg_stats['config']}条, 音频:{msg_stats['audio']}包({audio_bytes_total:,}字节), 控制:{msg_stats['control']}条",
            flush=True,
        )

    except websockets.ConnectionClosed:
        print(
            f"🔌 连接断开: {websocket.remote_address}, 剩余连接数: {len(websocket_users) - 1}",
            flush=True,
        )
    except websockets.InvalidState:
        print(f"⚠️ 连接状态异常: {websocket.remote_address}", flush=True)
    except Exception as e:
        print(f"❌ 连接异常: {websocket.remote_address} - {e}", flush=True)
    finally:
        # 无论什么原因断开连接，都要清理
        print(f"🧽 正在清理连接: {websocket.remote_address}", flush=True)
        await ws_reset(websocket)
        if websocket in websocket_users:
            websocket_users.remove(websocket)
            print(f"📊 当前剩余连接数: {len(websocket_users)}", flush=True)


async def async_vad(websocket: Any, audio_in: Any) -> tuple[int, int]:
    segments_result = model_manager.vad.generate(
        input=audio_in, **websocket.status_dict_vad
    )[0][
        "value"
    ]  # type: ignore
    # print(segments_result)

    speech_start = -1
    speech_end = -1

    if len(segments_result) == 0 or len(segments_result) > 1:  # type: ignore
        return speech_start, speech_end
    if segments_result[0][0] != -1:  # type: ignore
        speech_start = segments_result[0][0]  # type: ignore
    if segments_result[0][1] != -1:  # type: ignore
        speech_end = segments_result[0][1]  # type: ignore
    return speech_start, speech_end  # type: ignore


async def async_asr(websocket: Any, audio_in: Any):
    """
    离线ASR处理函数（异步）

    1. 处理音频数据
    2. 说话人识别
    3. 标点符号处理
    4. 生成响应
    """
    if len(audio_in) > 0:
        # print(len(audio_in))
        rec_result = model_manager.asr.generate(
            input=audio_in, **websocket.status_dict_asr
        )[
            0
        ]  # type: ignore
        # print("offline_asr, ", rec_result)

        # 标点符号处理 - 使用增强配置和更智能的条件判断
        if (
            CONFIG.punctuation.ENABLE_PUNC_OFFLINE
            and model_manager.punc is not None
            and len(str(rec_result.get("text", "")))
            >= CONFIG.punctuation.MIN_TEXT_LENGTH
            and len(str(rec_result.get("text", "")))
            <= CONFIG.punctuation.MAX_TEXT_LENGTH
        ):
            try:
                original_text = str(rec_result.get("text", ""))
                # print("offline, before punc", rec_result, "cache", websocket.status_dict_punc)
                punc_result = model_manager.punc.generate(
                    input=original_text, **websocket.status_dict_punc
                )[0]
                rec_result = punc_result
                # print("offline, after punc", rec_result)
                print(
                    f'📝 标点符号处理: "{original_text}" → "{punc_result.get("text", "")}"',
                    flush=True,
                )
            except Exception as e:
                print(f"⚠️ 离线标点符号处理错误: {e}")
                # 保持原始识别结果

        text_result = str(rec_result.get("text", ""))
        if len(text_result) > 0:
            mode = (
                CONFIG.processing.MODE_2PASS_OFFLINE
                if CONFIG.processing.MODE_2PASS in websocket.mode
                else websocket.mode
            )
            status = "partial"
            confidence = (
                float(rec_result.get("confidence", 0.95))
                if rec_result.get("confidence") is not None
                else 0.95
            )

            # 使用ResponseManager创建响应
            response_data = response_manager.success(
                text=text_result,
                mode=mode,
                status=status,
                wav_name=websocket.wav_name,
                speaker_info=None,
                confidence=confidence,
                duration=3000,
            )

            # 转换为JSON格式发送
            message = json.dumps(response_data.to_dict())

            # 生成日志信息
            log_msg = response_manager.get_log_message(text_result, None, mode, status)

            print(log_msg, flush=True)

            # 向客户端发送响应
            await websocket.send(message)

    else:
        mode = (
            CONFIG.processing.MODE_2PASS_OFFLINE
            if CONFIG.processing.MODE_2PASS in websocket.mode
            else websocket.mode
        )
        status = "partial"

        # 使用ResponseManager创建空响应
        response_data = response_manager.success(
            text="",
            mode=mode,
            status=status,
            wav_name=websocket.wav_name,
            speaker_info=None,
            confidence=0.0,
            duration=0,
        )

        message = json.dumps(response_data.to_dict())
        print(f"📤 发送空识别结果 → 客户端 (模式:{mode})", flush=True)
        await websocket.send(message)


async def async_asr_online(websocket: Any, audio_in: Any) -> None:
    """
    在线ASR处理函数（异步）

    1. 处理音频数据
    2. 说话人识别
    3. 标点符号处理
    4. 生成响应
    """
    if len(audio_in) > 0:
        # print(websocket.status_dict_asr_online.get("is_final", False))
        rec_result = model_manager.asr_streaming.generate(  # type: ignore
            input=audio_in, **websocket.status_dict_asr_online
        )[0]
        # print("online, ", rec_result)
        if (
            websocket.mode == CONFIG.processing.MODE_2PASS
            and websocket.status_dict_asr_online.get("is_final", False)
        ):
            return
            #     websocket.status_dict_asr_online["cache"] = dict()

        text_result = str(rec_result.get("text", ""))
        if len(text_result) > 0:
            is_final = websocket.status_dict_asr_online.get("is_final", False)
            mode = (
                CONFIG.processing.MODE_2PASS_ONLINE
                if CONFIG.processing.MODE_2PASS in websocket.mode
                else websocket.mode
            )
            status = "final" if is_final else "partial"
            confidence = (
                float(rec_result.get("confidence", 0.85))
                if rec_result.get("confidence") is not None
                else 0.85
            )

            # 使用ResponseManager创建响应
            response_data = response_manager.success(
                text=text_result,
                mode=mode,
                status=status,
                wav_name=websocket.wav_name,
                speaker_info=None,
                confidence=confidence,
                duration=1000,  # 在线模式通常是较短的音频段
            )

            # 转换为JSON格式发送
            message = json.dumps(response_data.to_dict())

            # 生成日志信息
            log_msg = response_manager.get_log_message(text_result, None, mode, status)
            print(log_msg, flush=True)

            await websocket.send(message)


async def main():
    if len(args.certfile) > 0:
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ssl_cert = args.certfile
        ssl_key = args.keyfile
        ssl_context.load_cert_chain(ssl_cert, keyfile=ssl_key)

        async with websockets.serve(
            ws_serve,
            args.host,
            args.port,
            subprotocols=CONFIG.server.WEBSOCKET_SUBPROTOCOLS,  # type: ignore
            # ping_interval=CONFIG.server.PING_INTERVAL,
            ping_interval=None,  # 禁用 心跳机制
            ssl=ssl_context,
        ):
            print(f"🚀 语音识别 WebSocket 服务已启动在 wss://{args.host}:{args.port}")
            print("等待连接...")

            await asyncio.Future()  # run forever
    else:
        async with websockets.serve(
            ws_serve,
            args.host,
            args.port,
            subprotocols=CONFIG.server.WEBSOCKET_SUBPROTOCOLS,  # type: ignore
            # ping_interval=CONFIG.server.PING_INTERVAL,
            ping_interval=None,  # 禁用 心跳机制
        ):
            print(f"🚀 语音识别 WebSocket 服务已启动在 ws://{args.host}:{args.port}")
            print("等待连接...")
            await asyncio.Future()  # run forever


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n语音识别服务器已停止!")
